package com.saic.data.job;

import java.util.UUID;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.log4j.Logger;

import com.saic.data.dotask.DoTask;
import com.saic.data.util.FileUtil;

/**
 * @DESC:写锁
 * @author Tansheng
 */
public class WriteJob {

	private String projectName = "finance_platform";
	private String zookeeperAddress = "10.32.47.104:2181";
	private CuratorFramework client;
	private String jobID = UUID.randomUUID().toString();
	private static Logger logger = Logger.getLogger(WriteJob.class);
	private String writeLockPath = "/locks/writeLock";

	public WriteJob() throws Exception {

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
		this.client = CuratorFrameworkFactory.builder()
				.connectString(this.zookeeperAddress).retryPolicy(retryPolicy)
				.namespace(this.projectName).build();
		client.start();

	}

	public void doWriteJob(String fileName, DoTask doTask) throws Exception {
		try {
			String tmp = getFTPFileName(fileName);
			if (tmp != null && !tmp.trim().equals("")) {
				if (this.getWriteLock(tmp)) {
					logger.info("Started job for file: " + fileName);
					doTask.doTask();
				} else {
					logger.info("Found job exists for file: " + fileName);
					this.client.close();
				}
			}
		} catch (Exception e) {
			throw e;
		} finally {}
	}

	private boolean getWriteLock(String _ftpFileName) throws Exception {
		return GetLock.getLock(this.client, this.jobID, this.writeLockPath,
				_ftpFileName);
	}
	private void dropWriteLock(String _ftpFileName) throws Exception {
		GetLock.dropLock(this.client, this.writeLockPath,_ftpFileName);
	}

	private String getFTPFileName(String fileName) {
		return FileUtil.getMD5(fileName);
	}

	public void close() {
		if (null != this.client)
			this.client.close();
	}

}
